package com.daydreamdev.secondskill.service.impl;

import com.daydreamdev.secondskill.common.StockWithRedis.RedisKeysConstant;
import com.daydreamdev.secondskill.common.StockWithRedis.StockWithRedis;
import com.daydreamdev.secondskill.common.utils.CacheKey;
import com.daydreamdev.secondskill.common.utils.RedisPool;
import com.daydreamdev.secondskill.common.utils.RedisPoolUtil;
import com.daydreamdev.secondskill.dao.StockOrderMapper;
import com.daydreamdev.secondskill.dao.UserMapper;
import com.daydreamdev.secondskill.pojo.Stock;
import com.daydreamdev.secondskill.pojo.StockOrder;
import com.daydreamdev.secondskill.pojo.User;
import com.daydreamdev.secondskill.service.api.OrderService;
import com.daydreamdev.secondskill.service.api.StockService;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

import java.util.List;

/**
 * @auther G.Fukang
 * @date 6/7 12:44
 */
@Slf4j
@Transactional(rollbackFor = Exception.class)
@Service //(value = "OrderService")
public class OrderServiceImpl implements OrderService {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private StockService stockService;

    @Autowired
    private StockOrderMapper orderMapper;

    @Autowired
    private UserMapper userMapper;

    @Override
    public int createWrongOrder(int sid) {
        //校验库存
        Stock stock = checkStock(sid);
        //扣库存
        saleStock(stock);
        //创建订单
        int id = createOrder(stock);
        return id;
    }

    @Override
    public int createOptimisticOrder(int sid) {
        //校验库存
        Stock stock = checkStock(sid);
        //乐观锁更新库存
        boolean success = saleStockOptimistic(stock);
        if (!success) {
            throw new RuntimeException("过期库存值，更新失败");
        }
        //创建订单
        createOrder(stock);
        return stock.getCount() - (stock.getSale() + 1);
    }

    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
    @Override
    public int createPessimisticOrder(int sid) {
        //校验库存(悲观锁for update)
        Stock stock = checkStockForUpdate(sid);
        //更新库存 乐观锁
        saleStockOptimistic(stock);
        //创建订单
        createOrder(stock);
//        int res = stock.getCount().intValue() - stock.getSale().intValue();
        int res = stock.getCount().intValue() - 1;
        return res;
    }

    @Override
    public int createVerifiedOrder(Integer sid, Integer userId, String verifyHash) throws Exception {

        // 验证是否在抢购时间内
        log.info("请自行验证是否在抢购时间内,假设此处验证成功");

        // 验证hash值合法性
        String hashKey = CacheKey.HASH_KEY.getKey() + "_" + sid + "_" + userId;
        System.out.println(hashKey);
        String verifyHashInRedis = stringRedisTemplate.opsForValue().get(hashKey);
        if (!verifyHash.equals(verifyHashInRedis)) {
            throw new Exception("hash值与Redis中不符合");
        }
        log.info("验证hash值合法性成功");

        // 检查用户合法性
        User user = userMapper.selectByPrimaryKey(userId.longValue());
        if (user == null) {
            throw new Exception("用户不存在");
        }
        log.info("用户信息验证成功：[{}]", user.toString());

        // 检查商品合法性
        Stock stock = stockService.getStockById(sid);
        if (stock == null) {
            throw new Exception("商品不存在");
        }
        log.info("商品信息验证成功：[{}]", stock.toString());

        //乐观锁更新库存
        boolean success = saleStockOptimistic(stock);
        if (!success) {
            throw new RuntimeException("过期库存值，更新失败");
        }
        log.info("乐观锁更新库存成功");

        //创建订单
        createOrderWithUserInfoInDB(stock, userId);
        log.info("创建订单成功");

        return stock.getCount() - (stock.getSale() + 1);
    }

    @Override
    public void createOrderByMq(Integer sid, Integer userId) throws Exception {

        // 模拟多个用户同时抢购，导致消息队列排队等候10秒
        Thread.sleep(10000);

        Stock stock;
        //校验库存（不要学我在trycatch中做逻辑处理，这样是不优雅的。这里这样处理是为了兼容之前的秒杀系统文章）
        try {
            stock = checkStock(sid);
        } catch (Exception e) {
            log.info("库存不足！");
            return;
        }
        //乐观锁更新库存
        boolean updateStock = saleStockOptimistic(stock);
        if (!updateStock) {
            log.warn("扣减库存失败，库存已经为0");
            return;
        }

//        log.info("扣减库存成功，剩余库存：[{}]", stock.getCount() - stock.getSale() - 1);
        log.info("扣减库存成功，剩余库存：[{}]", stock.getCount() - stock.getSale());
        stockService.delStockCountCache(sid);
        log.info("删除库存缓存");

        //创建订单
        log.info("写入订单至数据库");
        createOrderWithUserInfoInDB(stock, userId);
        log.info("写入订单至缓存供查询");
        createOrderWithUserInfoInCache(stock, userId);
        log.info("下单完成");

    }

    @Override
    public Boolean checkUserOrderInfoInCache(Integer sid, Integer userId) throws Exception {
        String key = CacheKey.USER_HAS_ORDER.getKey() + "_" + sid;
        log.info("检查用户Id：[{}] 是否抢购过商品Id：[{}] 检查Key：[{}]", userId, sid, key);
        return stringRedisTemplate.opsForSet().isMember(key, userId.toString());
    }

    /**
     * 检查库存
     *
     * @param sid
     * @return
     */
    private Stock checkStock(int sid) {
        Stock stock = stockService.getStockById(sid);
        if (stock.getSale().equals(stock.getCount())) {
            throw new RuntimeException("库存不足");
        }
        return stock;
    }

    /**
     * 检查库存 ForUpdate
     *
     * @param sid
     * @return
     */
    private Stock checkStockForUpdate(int sid) {
        Stock stock = stockService.getStockByIdForUpdate(sid);
        if (stock.getSale().equals(stock.getCount())) {
            throw new RuntimeException("库存不足");
        }
        return stock;
    }

    /**
     * 更新库存
     *
     * @param stock
     */
    private void saleStock(Stock stock) {
        stock.setSale(stock.getSale() + 1);
        stockService.updateStockById(stock);
    }

    /**
     * 更新库存 乐观锁
     *
     * @param stock
     */
    private boolean saleStockOptimistic(Stock stock) {
        log.info("-v3-3--查询数据库，尝试更新库存 - saleStockOptimistic");
        int count = stockService.updateStockByOptimistic(stock);
        return count != 0;
    }

    /**
     * 创建订单
     *
     * @param stock
     * @return
     */
    private int createOrder(Stock stock) {
        StockOrder order = new StockOrder();
        order.setSid(stock.getId());
        order.setName(stock.getName());
        return orderMapper.insertSelective(order);
    }

    /**
     * 创建订单：保存用户订单信息到数据库
     *
     * @param stock
     * @return
     */
    private int createOrderWithUserInfoInDB(Stock stock, Integer userId) {
        StockOrder order = new StockOrder();
        order.setSid(stock.getId());
        order.setName(stock.getName());
//        order.setUserId(userId);
        return orderMapper.insertSelective(order);
    }

    /**
     * 创建订单：保存用户订单信息到缓存
     *
     * @param stock
     * @return 返回添加的个数
     */
    private Long createOrderWithUserInfoInCache(Stock stock, Integer userId) {
        String key = CacheKey.USER_HAS_ORDER.getKey() + "_" + stock.getId().toString();
        log.info("写入用户订单数据Set：[{}] [{}]", key, userId.toString());
        return stringRedisTemplate.opsForSet().add(key, userId.toString());
    }


    //——————————————————————————————————————————————

    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
    @Override
    public int createOrderWithLimitAndRedis(int sid) throws Exception {
        // 校验库存，从 Redis 中获取
        Stock stock = checkStockWithRedis(sid);
        // 乐观锁更新库存和Redis
        saleStockOptimsticWithRedis(stock);
        // 创建订单
        int res = createOrder(stock);
        return res;
    }

    // Redis 中校验库存
    private Stock checkStockWithRedis(int sid) throws Exception {
        Integer count = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_COUNT + sid));
        Integer sale = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_SALE + sid));
        Integer version = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_VERSION + sid));
        if (count < 1) {
            //log.info("库存不足");
            throw new RuntimeException("库存不足 Redis currentCount: " + sale);
        }
        Stock stock = new Stock();
        stock.setId(sid);
        stock.setCount(count);
        stock.setSale(sale);
        stock.setVersion(version);
        // 此处应该是热更新，但是在数据库中只有一个商品，所以直接赋值
        stock.setName("手机");

        return stock;
    }

    // 更新 DB 和 Redis
    private void saleStockOptimsticWithRedis(Stock stock) throws Exception {
        int res = stockService.updateStockByOptimistic(stock);
        if (res == 0){
            throw new RuntimeException("并发更新库存失败") ;
        }
        // 更新 Redis
        StockWithRedis.updateStockWithRedis(stock);
    }
    // Redis 多个写入操作的事务
    public static void updateStockWithRedis(Stock stock) {
        Jedis jedis = null;
        try {
            jedis = RedisPool.getJedis();
            // 开始事务
            Transaction transaction = jedis.multi();
            // 事务操作
            RedisPoolUtil.decr(RedisKeysConstant.STOCK_COUNT + stock.getId());
            RedisPoolUtil.incr(RedisKeysConstant.STOCK_SALE + stock.getId());
            RedisPoolUtil.incr(RedisKeysConstant.STOCK_VERSION + stock.getId());
            // 结束事务
            List<Object> list = transaction.exec();
        } catch (Exception e) {
            //log.error("updateStock 获取 Jedis 实例失败：", e);
        } finally {
            RedisPool.jedisPoolClose(jedis);
        }
    }

    // Redis 校验库存
    private Stock checkStockWithRedisWithDel(int sid) throws Exception {
        Integer count = null;
        Integer sale = null;
        Integer version = null;
        List<String> data = RedisPoolUtil.listGet(RedisKeysConstant.STOCK + sid);
        if (data.size() == 0) {
            // Redis 不存在，先从数据库中获取，再放到 Redis 中
            Stock newStock = stockService.getStockById(sid);
            RedisPoolUtil.listPut(RedisKeysConstant.STOCK + newStock.getId(), String.valueOf(newStock.getCount()),
                    String.valueOf(newStock.getSale()), String.valueOf(newStock.getVersion()));
            count = newStock.getCount();
            sale = newStock.getSale();
            version = newStock.getVersion();
        } else {
            count = Integer.parseInt(data.get(0));
            sale = Integer.parseInt(data.get(1));
            version = Integer.parseInt(data.get(2));
        }
        if (count < 1) {
            //log.info("库存不足");
            throw new RuntimeException("库存不足 Redis currentCount: " + sale);
        }
        Stock stock = new Stock();
        stock.setId(sid);
        stock.setCount(count);
        stock.setSale(sale);
        stock.setVersion(version);
        // 此处应该是热更新，但是在数据库中只有一个商品，所以直接赋值
        stock.setName("手机");
        return stock;
    }
    private void saleStockOptimsticWithRedisWithDel(Stock stock) throws Exception {
        // 乐观锁更新数据库
        int res = stockService.updateStockByOptimistic(stock);
        // 删除缓存，应该使用 Redis 事务
        RedisPoolUtil.del(RedisKeysConstant.STOCK + stock.getId());
        //log.info("删除缓存成功");
        if (res == 0) {
            throw new RuntimeException("并发更新库存失败");
        }
    }

    //Kafka 异步


    @Autowired
    private KafkaTemplate kafkaTemplate;
    private Gson gson = new GsonBuilder().create();

    // 向 Kafka 发送消息
    @Override
    public void createOrderWithLimitAndRedisAndKafka(int sid) throws Exception {
        // 校验库存
        Stock stock = checkStockWithRedis(sid);
        // 下单请求发送至 kafka，需要序列化 stock
        String kafkaTopic = "SECONDS-KILL-TOPIC";
        kafkaTemplate.send(kafkaTopic, gson.toJson(stock));
        //log.info("消息发送至 Kafka 成功");
    }

    // Kafka 消费消息执行创建订单业务
    @Override
    public int consumerTopicToCreateOrderWithKafka(Stock stock) throws Exception {
        // 乐观锁更新库存和 Redis
        saleStockOptimsticWithRedis(stock);
        int res = createOrder(stock);
        if (res == 1) {
            log.info(">>>>>> Kafka 消费 Topic 创建订单成功");
        } else {
            log.info(">>>>>> Kafka 消费 Topic 创建订单失败");
        }
        return res;
    }



}
